package com.dtruth.dataflow.dao;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import com.alibaba.fastjson.JSONObject;
import com.dtruth.dataflow.dbmanager.HBaseManager;
import com.dtruth.dataflow.mq.RabbitMQProducer;

/**
 * 
 * @author 岳海亮
 * @email yhl@feheadline.com
 * @date 2014年10月27日
 */
public class HBaseDAO {

	private Configuration conf = null;

	public HBaseDAO() {
		conf = HBaseManager.getInstance().getConf();
	}

	public void writeRow(String tablename, String rowKey, String family,
			String qualifier, String value) {
		HTable table = null;
		try {
			table = new HTable(conf, tablename);
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier),
					Bytes.toBytes(value));
			table.put(put);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 
	 * @param tablename
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	public Map<String, Object> selectRow(String tableName, String rowKey)
			throws IOException {
		Map<String, Object> map = new HashMap<String, Object>();
		HTable table = new HTable(conf, tableName);
		Get g = new Get(Bytes.toBytes(rowKey));
		Result rs = table.get(g);
		for (Cell cell : rs.rawCells()) {
			String qualifier = new String(CellUtil.cloneQualifier(cell),
					"UTF-8");
			String value = new String(CellUtil.cloneValue(cell), "UTF-8");
			if (!qualifier.equals("$category")
					// && !qualifier.equals("referer")
					&& !qualifier.equals("spider")
					&& !qualifier.equals("drill_relation")
					&& !qualifier.equals("content"))
				// && !qualifier.equals("updated"))
				map.put(qualifier, value);
		}
		table.close();
		if (!map.isEmpty())
			map.put("key", rowKey);
		return map;
	}

	/**
	 * 
	 * @param tablename
	 * @param rowKey
	 * @return
	 */
	public void imgCounter(String rowKey) {
		Map<String, Object> map = new HashMap<String, Object>();
		HTable table = null;
		try {
			table = new HTable(conf, "crawled_bin");
			Get g = new Get(Bytes.toBytes(rowKey));
			Result rs = table.get(g);
			for (Cell cell : rs.rawCells()) {
				String qualifier = new String(CellUtil.cloneQualifier(cell));
				if (!qualifier.equals("file")) {
					String value = new String(CellUtil.cloneValue(cell));
					map.put(qualifier, value);
				} else {
					// ByteArrayInputStream bais = new
					// ByteArrayInputStream(CellUtil.cloneValue(cell));
					// BufferedImage bi1 =ImageIO.read(bais);
					// File w2 = new File("");//可以是jpg,png,gif格式
					// ImageIO.write(bi1, "jpg", w2);
				}

			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public Set<String> getImageUrls(String rowKey) {
		Set<String> set = new HashSet<String>();
		HTable table = null;
		try {
			table = new HTable(conf, "crawled_bin");
			Get g = new Get(Bytes.toBytes(rowKey));
			Result rs = table.get(g);
			for (Cell cell : rs.rawCells()) {
				String qualifier = new String(CellUtil.cloneQualifier(cell));
				if (qualifier.equals("url")) {

				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (table != null)
					table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return set;
	}

	public Set<String> scanCrawled(String tablename) {
		Set<String> set = new HashSet<String>();
		HTable table = null;
		try {
			table = new HTable(conf, tablename);
			Scan scan = new Scan();
			ResultScanner rs = table.getScanner(scan);
			Iterator<Result> it = rs.iterator();
			int i = 0;
			while (it.hasNext()) {
				Result result = it.next();
				Map<Integer, String> map = new TreeMap<Integer, String>();
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					if (qualifier.equals("urllib"))
						map.put(0, new String(CellUtil.cloneValue(cell)));
					if (qualifier.equals("category1"))
						map.put(1, new String(CellUtil.cloneValue(cell)));
					if (qualifier.equals("category2"))
						map.put(2, new String(CellUtil.cloneValue(cell)));
					if (qualifier.equals("category3"))
						map.put(3, new String(CellUtil.cloneValue(cell)));
				}
				String category = map.get(0) + "##" + map.get(1) + "##"
						+ map.get(2) + "##" + map.get(3);
				map.clear();
				set.add(category);
				if (i > 1000) {
					System.out.println(category);
					i = 0;
				}
				i++;
			}
			System.out.println(i);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (table != null)
					table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return set;
	}

	public Map<String, String> scan(String tablename, String qname, String data) {
		Map<String, String> map = new HashMap<String, String>();
		HTable table = null;
		try {
			table = new HTable(conf, tablename);
			Scan scan = new Scan();
			ResultScanner rs = table.getScanner(scan);
			Iterator<Result> it = rs.iterator();
			int i = 0;
			while (it.hasNext()) {
				Result result = it.next();
				String rowKey = new String(result.getRow());
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					if (qualifier.equals(qname)) {
						String value = new String(CellUtil.cloneValue(cell));
						if (value != null && !value.equals("")
								&& !value.equals("[]")) {
							if (value.indexOf(data) > -1)
								map.put(rowKey, value);

						}
					}
				}
				if (map.keySet().size() > 2)
					return map;

			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (table != null)
					table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return null;
	}

	public Set<String> scanCrawledBin(String tablename) {
		Set<String> set = new HashSet<String>();
		HTable table = null;
		try {
			table = new HTable(conf, tablename);
			Scan scan = new Scan();
			ResultScanner rs = table.getScanner(scan);
			Iterator<Result> it = rs.iterator();
			while (it.hasNext()) {
				Result result = it.next();
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					if (qualifier.equals("url")) {
						String value = new String(CellUtil.cloneValue(cell));
						System.out.println(value);
						if (!value.endsWith(".pdf") && !value.endsWith(".PDF")) {
							String row = new String(CellUtil.cloneRow(cell));
							System.out.println(row);
							set.add(row);
							break;
						}
					}
					if (set.size() > 0)
						break;

				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (table != null)
					table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return set;
	}

	public Set<String> scanQualifier(String tablename, String q) {
		Set<String> set = new HashSet<String>();
		HTable table = null;
		try {
			table = new HTable(conf, tablename);
			Scan scan = new Scan();
			ResultScanner rs = table.getScanner(scan);
			Iterator<Result> it = rs.iterator();
			int i = 0;
			while (it.hasNext()) {
				Result result = it.next();
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					if (q == null)
						set.add(qualifier);
					else if (qualifier.equals(q))
						set.add(new String(CellUtil.cloneValue(cell)));
				}
				if (i > 1000) {
					System.out.println(set);
					i = 0;
				}
				i++;
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (table != null)
					table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return set;
	}

	/**
	 * 
	 * @param tableName
	 * @return
	 */
	public void pushResdis(String tableName, long starttime, long endtime) {
		try {
			RedisDAO rc = new RedisDAO("tree", 6379, 2);
			HConnection hconn = HConnectionManager.createConnection(conf);
			HTableInterface table = hconn.getTable(tableName);
			ResultScanner rs = table.getScanner(new Scan());
			Iterator<Result> it = rs.iterator();
			int i = 0;
			int k = 0;
			int count = 0;
			while (it.hasNext()) {
				Result result = it.next();
				String time = null;
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					if (qualifier.equals("updated")) {
						time = new String(CellUtil.cloneValue(cell));
					}
				}
				String md5 = new String(result.getRow());
				if (Long.valueOf(time) >= starttime
						&& Long.valueOf(time) < endtime) {
					rc.rpush(md5);
					count++;
					System.out.println("push md5:" + md5);
				}
				if (i > 1000) {
					System.out.println(k + ":" + md5);
					i = 0;
					k++;
				}
				i++;
			}
			System.out.println("push count:" + count);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 
	 * @param tableName
	 * @return
	 */
	public void push2RabbitMQ(String tableName, long starttime, long endtime) {
		try {
			HConnection hconn = HConnectionManager.createConnection(conf);
			HTableInterface table = hconn.getTable(tableName);
			ResultScanner rs = table.getScanner(new Scan());
			Iterator<Result> it = rs.iterator();
			int count = 0;
			JSONObject jobj = new JSONObject();
			while (it.hasNext()) {
				Result result = it.next();
				String time = null;
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					String value = new String(CellUtil.cloneValue(cell));
					jobj.put(qualifier, value);
					if (qualifier.equals("updated")) {
						time = new String(CellUtil.cloneValue(cell));
					}

				}
				String md5 = new String(result.getRow());
				if (Long.valueOf(time) >= starttime
						&& Long.valueOf(time) < endtime) {
					jobj.put("key", md5);
					RabbitMQProducer.instance.sendMessage(jobj.toJSONString());
					count++;
					System.out.println("push md5:" + md5);
				}
				if (count > 20) {
					break;
				}
			}
			System.out.println("push count:" + count);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public Map<String, String> getBody(int max) {
		Map<String, String> map = new HashMap<String, String>();
		try {
			HConnection hconn = HConnectionManager.createConnection(conf);
			HTableInterface table = hconn.getTable("crawled");
			ResultScanner rs = table.getScanner(new Scan());
			Iterator<Result> it = rs.iterator();
			while (it.hasNext()) {
				Result result = it.next();
				String url = null;
				String body = null;
				for (Cell cell : result.rawCells()) {
					String qualifier = new String(CellUtil.cloneQualifier(cell));
					if (qualifier.equals("url"))
						url = new String(CellUtil.cloneValue(cell));
					if (qualifier.equals("body"))
						body = new String(CellUtil.cloneValue(cell));
				}
				if (url != null && body != null)
					map.put(url, body);
				if (map.keySet().size() > max)
					break;
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return map;
	}

	public Map<String, Object> getByColNames(String tableName, String rowKey,
			String... colNames) throws IOException {
		Map<String, Object> map = new HashMap<String, Object>();
		HTable table = new HTable(conf, tableName);
		Get g = new Get(Bytes.toBytes(rowKey));
		Result rs = table.get(g);
		for (Cell cell : rs.rawCells()) {
			String qualifier = new String(CellUtil.cloneQualifier(cell));
			for (String colName : colNames)
				if (qualifier.equals(colName))
					map.put(colName, new String(CellUtil.cloneValue(cell)));
		}
		table.close();
		return map;
	}

	public boolean createTable(String tableName, String[] families) {
		HBaseAdmin hadmin = null;
		try {
			// table create,disable,exist ,drop,use HBaseAdmin
			hadmin = new HBaseAdmin(conf);
			if (hadmin.tableExists(tableName)) {
				hadmin.disableTable(tableName);
				hadmin.deleteTable(tableName);
				System.out.println("table " + tableName + " exist,delete it.");

			}
			TableName tn = TableName.valueOf(tableName);
			HTableDescriptor tbdesc = new HTableDescriptor(tn);
			for (String family : families) {
				tbdesc.addFamily(new HColumnDescriptor(family));
			}
			hadmin.createTable(tbdesc);
		} catch (MasterNotRunningException e) {
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				if (hadmin != null)
					hadmin.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		System.out.println("table " + tableName + " create ok.");
		return true;
	}
}
